/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.IOException;
import java.io.LineNumberReader;
import java.io.Serializable;
import java.io.StringReader;
import java.util.HashMap;
import java.util.Map;

import org.msgpack.annotation.Message;
import org.sss.mapreduce.datatype.Packable;

@SuppressWarnings("serial")
@Message
public class Center implements Packable, Serializable {
  private static long SIZE = Long.SIZE / 8 + Double.SIZE / 8 + Double.SIZE / 8;

	static final int xrange = 100;
	static final int yrange = 100;
	static final double epsilon = 0.000000001;
	
	long num;
	double x;
	double y;
	
	public Center() {}
	public Center(double x, double y) {
		this.x = x;
		this.y = y;
		this.num = 1;
	}
	public Center(String line) {
    String [] tokens = line.trim().split("\\s+");
		this.x = Double.parseDouble(tokens[0]);
		this.y = Double.parseDouble(tokens[1]);
		this.num = Long.parseLong(tokens[2]);
	}
	
	static Center getRandom() {
		Center tmp = new Center();
		tmp.x = Math.random() * xrange;
		tmp.y = Math.random() * yrange;
		return tmp;
	}

	public double distance(double x, double y) {
		return Math.sqrt((this.x - x) * (this.x - x) + (this.y - y) * (this.y - y));
	}

	private boolean compare(Center center) {
		if (num != center.num)
			return false;
		if ((x - center.x < epsilon) && (y - center.y < epsilon))
			return true;
		return false;
	}

	@Override
	public String toString() {
		return String.format("%f %f %d", x, y, num);
	}

	public void vote(double x, double y) { //accumulate
		this.num++;
		this.x += x;
		this.y += y;
	}

	public void weightedVote(Center c0) { //accumulate
		this.num += c0.num;
		this.x += c0.x * c0.num;
		this.y += c0.y * c0.num;
	}
	
	public void normalize() {// to be center
		this.x /= this.num;
		this.y /= this.num;
	}
	
	static Map<Integer, Center>parse(String s) throws IOException {
		Map<Integer, Center>centers = new HashMap<Integer, Center>();
		LineNumberReader lnr	= new LineNumberReader(new StringReader(s));		
		String line;
		while ((line = lnr.readLine()) != null) {
      String [] tokens = line.trim().split("\\s+");
			centers.put(Integer.parseInt(tokens[0]), new Center(tokens[1]));
		}
		return centers;
	}

	static Map<Integer, Center>createInitial(int numOfCenters) {
		Map<Integer, Center>l = new HashMap<Integer, Center>();
		for (int i = 0; i < numOfCenters; i++)  
			l.put(i, getRandom());
		return l;
	}
	
	static String arrayStr(Map<Integer, Center>a0) {
		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < a0.size(); i++)  
			sb.append(String.format("%d\t%s\n", i, a0.get(i)));
		return sb.toString();
	}
	
	
	static boolean compare(Map<Integer, Center>a0, Map<Integer, Center>a1) {
		if (a0.size() != a1.size()) {
			return false;
    }

		for (int i = 0; i < a0.size(); i++) {
			if (! a0.get(i).compare(a1.get(i))) {
				return false;
      }
    }
		return true;	
	}
	
	public static boolean compare(Center[] a0, Center[] a1) {
		if (a0.length != a1.length)
			return false;
		for (int i = 0; i < a0.length; i++) 
			if (! a0[i].compare(a1[i]))
				return false;
		return true;
	}

  @Override
  public int slot() {
    return this.hashCode();
  }

  @Override
  public long getRoughSize() {
    return SIZE;
  }
}
